// JobPool.cpp: implementation of the CJobPool class.
//Author : Pradeep K Sahu
//Email : sahupk@yahoo.com
//Date  :29-Mar-2003
//Note   : You are free to use the code, and modify. Provided you
//         don't remove this comments. It will be great if you can provide
//		   your feedback.
//Last Modified : 26-May-2003.
//Change History :
// 26-May-2003 : memory leak solved. Which was detected and also solved by Guitool.Thanks to him. 
//////////////////////////////////////////////////////////////////////

#include "stdafx.h"
#include "JobQ.h"
#include "MThreadedJobQ.h"
static UINT REFRESH_LIST = ::RegisterWindowMessage("REFRESH LIST");

#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////


CMThreadedJobQ::CMThreadedJobQ()
{
	m_MaxNoOfExecuter = 2;
	m_pause = FALSE;
	m_pObserverThread = AfxBeginThread(JobObserverThreadFunction,this);
	m_pFreeEList =NULL;
	m_NoOfExecuter =0;
	m_pAllEList = NULL;
}


CMThreadedJobQ::~CMThreadedJobQ()
{
	//The lines below are added later to solve the memory leak.
	//The leaks are found and solved by "Guitool" having member no Member No. 7253 
	THNODE* pTempNode;
	while (m_pAllEList != NULL) 
	{	
		pTempNode = m_pAllEList->pNext;
		delete m_pAllEList->pExecuter;		
		delete m_pAllEList;		
		m_pAllEList = pTempNode;	
	}	

	while (m_pFreeEList != NULL) 
	{	pTempNode = m_pFreeEList->pNext;		
		delete m_pFreeEList;		
		m_pFreeEList = pTempNode;	
	}	

	m_pObserverThread->ExitInstance();	
	delete m_pObserverThread;
}


void CMThreadedJobQ::pause()
{
	this->m_pause = TRUE;
}

void CMThreadedJobQ::resume()
{
	this->m_pause = FALSE;
	this->m_pObserverThread->ResumeThread();
}


UINT CMThreadedJobQ::JobObserverThreadFunction(LPVOID pParam)
{
	CMThreadedJobQ *pMTJQ = (CMThreadedJobQ *)pParam;
	CJobExecuter *pJExecuter;
	TRACE0("Createing JobObserver Thread");
	while(TRUE)
	{
		Sleep(100); //This line can be removed..
		if(pMTJQ->m_pause != TRUE)
		{
			while(!pMTJQ->m_jobQList.IsEmpty() )
			{
				//Get Free Executer
				pJExecuter = pMTJQ->getJobExecuter();
				if( pJExecuter!=NULL)
				{				
					pMTJQ->m_cs.Lock();
					pJExecuter->execute(pMTJQ->m_jobQList.GetHead());
					pMTJQ->m_jobQList.RemoveHead();
					AfxGetApp()->m_pMainWnd->PostMessage(REFRESH_LIST);
					pMTJQ->m_cs.Unlock();
				}
				else
				{
					break;
				}
				if(pMTJQ->m_pause == TRUE)
					break;
			}
		}
		pMTJQ->m_pObserverThread->SuspendThread();
	}
	return 0;
}

int CMThreadedJobQ::getNoOfExecuter()
{
	return this->m_NoOfExecuter;
}


int CMThreadedJobQ::getMaxNoOfExecuter()
{
	return this->m_MaxNoOfExecuter;
}


void CMThreadedJobQ::addJob(CJob *pJob)
{
	CJob * pTempJob;
	CSingleLock sLock(&this->m_cs);
	sLock.Lock();	
	POSITION pos,lastPos;
	pos = this->m_jobQList.GetHeadPosition();	
	lastPos = pos;
	if(pos != NULL)
		pTempJob =this->m_jobQList.GetHead();
	while(pos != NULL )
	{		
		if( pJob->getPriority() > pTempJob->getPriority())
			break;
		lastPos = pos;
		pTempJob = 	this->m_jobQList.GetNext(pos);		
	}	
	if(pos == NULL)	
		this->m_jobQList.AddTail(pJob);
	else
		this->m_jobQList.InsertBefore(lastPos,pJob);
	this->m_pObserverThread->ResumeThread();
	sLock.Unlock();
}

void CMThreadedJobQ::addFreeJobExecuter(CJobExecuter *pEx)
{
	m_cs.Lock();
	THNODE* node = new THNODE;
	node->pExecuter = pEx;
	node->pNext = this->m_pFreeEList;
	this->m_pFreeEList = node;
	m_cs.Unlock();
}

CJobExecuter* CMThreadedJobQ::getJobExecuter()
{
	THNODE *pTemp;
	CJobExecuter *pEx=NULL;
	m_cs.Lock();

	//Check if there is any free thread then return
	if(this->m_pFreeEList != NULL)
	{
		pTemp = this->m_pFreeEList;
		this->m_pFreeEList = this->m_pFreeEList->pNext;
		pEx = pTemp->pExecuter;
		delete pTemp ;
		m_cs.Unlock();
		return pEx;
	}

	if(this->m_NoOfExecuter < this->m_MaxNoOfExecuter)
	{
		//create a new thread and return the same
		pEx =  new CJobExecuter(this);
		this->addJobExecuter(pEx);
		this->m_NoOfExecuter++;
		m_cs.Unlock();
		return pEx;
	}
	m_cs.Unlock();
	return NULL;
}

void CMThreadedJobQ::addJobExecuter(CJobExecuter *pEx)
{
	m_cs.Lock();
	THNODE* node = new THNODE;
	node->pExecuter= pEx;
	node->pNext = this->m_pAllEList;
	this->m_pAllEList = node;
	m_cs.Unlock();
}

void CMThreadedJobQ::setMaxNoOfExecuter(int value)
{
	this->m_cs.Lock();
	if(value >1 && value <11)
		this->m_MaxNoOfExecuter = value;
	m_pObserverThread->ResumeThread();
	this->m_cs.Unlock();
}

void CMThreadedJobQ::deleteJobExecuter(CJobExecuter *pEx)
{
	THNODE* pNode,*pNodeP;
	CSingleLock singleLock(&m_cs);	
	singleLock.Lock();
	//Remove the node from the link list which contain the pEx  thread		
	if(this->m_pAllEList != NULL)
	{
		pNode = this->m_pAllEList;
		//if the first node contains the Executer  as pEx
		if(pNode->pExecuter == pEx )	
		{
		  this->m_pAllEList = pNode->pNext;
		  delete pNode;		  
		}
		else
		{
			//IF the first node does not contain the thread to be deleted then 
			//search and delete the same
			pNodeP =pNode;
			pNode  = pNode->pNext ;			
			while(pNode != NULL )
			{
				if(pNode->pExecuter== pEx ) break;
				pNodeP = pNode;
				pNode  = pNode->pNext ;			
			}//end while
			if(pNode!= NULL)
			{
				//remove the current node 
				pNodeP->pNext = pNode->pNext;
				delete pNode;
			}

		}//end else
	}//End IF
	this->m_NoOfExecuter--;
	singleLock.Unlock();
	pEx->stop();
	Sleep(1);
	delete pEx;
}
